package com.niezhiliang.springboot.rabbitmq.common.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.niezhiliang.springboot.rabbitmq.common.annotation.MQCutHandler;
import com.niezhiliang.springboot.rabbitmq.common.domain.BrokerMessageLog;
import com.niezhiliang.springboot.rabbitmq.common.domain.BrokerMessageLogExample;
import com.niezhiliang.springboot.rabbitmq.common.mapper.BrokerMessageLogMapper;
import com.rabbitmq.client.Channel;
import lombok.Data;
import lombok.NonNull;
import lombok.experimental.Accessors;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.io.Serializable;
import java.nio.charset.Charset;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author Cody
 * @Description: rabbitMq消息发送
 * @date 2019/3/29 14:42
 */
@Component
@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
public class MessageUtils
{
    private static final Logger LOGGER = Logger.getLogger(MessageUtils.class);
    @Autowired(required = false)
    private RabbitTemplate rabbitTemplate;

    private static BrokerMessageLogMapper brokerMessageLogMapper;

    @Autowired
    public void setBrokerMessageLogMapper(BrokerMessageLogMapper brokerMessageLogMapper) {
        MessageUtils.brokerMessageLogMapper = brokerMessageLogMapper;
    }

    public void sendMsg(String exchange, String routingKey, String method, String id, Object object){
        sendMsg(exchange,routingKey,method,id,object,true);
    }
    /**
     * 统一发送消息入口
     *
     * @param exchange   交换机
     * @param routingKey 匹配的key
     * @param method     下游执行的类中的方法名
     * @param id         全局MsgId, 消息主键，唯一标识
     * @param object     消息实体
     * @param isSaveMsg 是否把消息保存在数据库, 默认true
     */
    public void sendMsg(String exchange, String routingKey, String method, String id, Object object, boolean isSaveMsg)
    {

        if (object == null)
        {
            LOGGER.cody().warn("rabbitMQ消息为空，中止发送");
            return;
        }


        if (method == null)
        {
            method = "handler";
        }
        QueueMsgEntity msg;
        if (object instanceof QueueMsgEntity)
        {
            msg = ((QueueMsgEntity) object).setId(id).setMethod(method);
        }
        else
        {
            msg = new QueueMsgEntity().setObject(object).setId(id).setMethod(method);
        }
        if (isSaveMsg)
        {
            if (!insertMessageLog(msg.getId(), exchange, routingKey, method, JSONObject.toJSONString(msg.getObject())))
            {
                LOGGER.cody().error("rabbitMQ消息未发出，消息落库失败");
                return;
            }
        }
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(id);
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.convertAndSend(exchange, routingKey, JSONObject.toJSONString(msg), correlationData);


    }

    private boolean insertMessageLog(String id, String exchange, String routingKey, String method, String message)
    {
        //插入rabbitmq投递信息日志表
        BrokerMessageLog brokerMessageLog = new BrokerMessageLog();
        brokerMessageLog.setMessage_id(id);
        brokerMessageLog.setMessage(message);
        brokerMessageLog.setStatus(MsgStatus.SENDING.name());
        brokerMessageLog.setCreate_time(new Date()).setUpdate_time(new Date());
        brokerMessageLog.setMethod(method);
        brokerMessageLog.setExchange(exchange);
        brokerMessageLog.setRouting_key(routingKey);
        //下一次投递时间
        brokerMessageLog.setNext_retry(new Date(System.currentTimeMillis() + 1 * 60000));
        brokerMessageLog.setTry_count(0);
        brokerMessageLog.setDisabled(0);
        return brokerMessageLogMapper.insertSelective(brokerMessageLog) > 0;
    }

    /**
     * ack 手动确认模式
     */
    public interface AckMessageListener  extends ChannelAwareMessageListener
    {

        /**
         * ack模式不需要指定方法，自动会调用onMessage方法
         * 接到ack消息
         * springboot项目会调用此方法
         *
         * @param message
         * @param channel
         * @throws Exception
         */
        @RabbitHandler
        default void onMessage(@Payload String message, @Headers Map<String, Object> properties, Channel channel) throws IOException
        {
            long deliveryTag = (Long) properties.get(AmqpHeaders.DELIVERY_TAG);
            handlerMessage(message, deliveryTag, channel);
        }

        /**
         * 非springboot项目会调用此方法
         *
         * @param message
         * @param channel
         * @throws Exception
         */
        @Override
        default void onMessage(Message message, Channel channel) throws Exception
        {
            handlerMessage(new String(message.getBody(), Charset.defaultCharset()), message.getMessageProperties().getDeliveryTag(), channel);
        }

        /**
         * 消息统一处理
         *
         * @param msg
         * @param deliveryTag
         * @param channel
         * @throws IOException
         */
        default void handlerMessage(String msg, long deliveryTag, Channel channel) throws IOException
        {
            LOGGER.cody().debug("ACK-消息队列已收到消息，并进入公用onMessage，将根据method字段查找方法并尝试执行");
            LOGGER.cody().debug("ACK-消息队列已收到消息，message:{}", msg);

            /**
             *  在最前面最认，为防止业务处理时异常，而中断确认消息，导致消息队列堵死  确认消息
             * 消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
             * 不写channel.basicAck(deliveryTag,false)会自动确认
             * */
            confirmMsg(channel, deliveryTag);


            QueueMsgEntity msgObj = null;
            try
            {
                if (msg.trim().startsWith("["))
                {
                    msgObj = JSON.parseArray(msg).getObject(0, QueueMsgEntity.class);
                }
                else
                {
                    msgObj = JSONObject.parseObject(msg, QueueMsgEntity.class);
                }
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }

            String methodName = msgObj.getMethod();
            if (null == msgObj.getObject())
            {
                LOGGER.cody().warn("消息内容为null，跳过执行");
                confirmMsg(channel, deliveryTag);
                return;
            }
            java.lang.reflect.Method[] methods = ReflectionUtils.getAllDeclaredMethods(this.getClass());
            String finalMethodName = methodName;
            java.lang.reflect.Method method = Arrays.stream(methods).filter(m -> m.getName().equals(finalMethodName)).findFirst().orElse(null);
            if (method == null)
            {
                LOGGER.cody().warn("下流消费端没有找到指定的方法,跳过执行");
                confirmMsg(channel, deliveryTag);
                return;
            }
            Class<?>[] parameterTypes = method.getParameterTypes();
            Object data;
            if (Collection.class.isAssignableFrom(parameterTypes[0]))
            {
                if (msgObj.getObject() instanceof String)
                {
                    data = JSONArray.parseArray((String) msgObj.getObject(), parameterTypes[0]);
                }
                else
                {
                    data = JSONArray.toJavaObject((JSONArray) msgObj.getObject(), parameterTypes[0]);
                }
            }
            else
            {
                if (msgObj.getObject() instanceof String)
                {
                    data = JSONObject.parseObject((String) msgObj.getObject(), parameterTypes[0]);
                }
                else
                {
                    data = JSONObject.toJavaObject((JSONObject) msgObj.getObject(), parameterTypes[0]);
                }
            }
            /**整理入参  需要添加参数在此处加入*/
            List<Object> objs = Stream.of(data, deliveryTag, channel).collect(Collectors.toList());
            Object[] params = new Object[parameterTypes.length];
            for (int i = 0; i < parameterTypes.length; i++)
            {
                params[i] = objs.get(i);
            }
            /**这里注意:如果反射的方法抛出异常，会导致消息始终无法消费*/
            ReflectionUtils.invokeMethod(method, this, params);

        }


        /**
         * 确认消息，表示消息正确消费
         */
        default void confirmMsg(Channel channel, long deliveryTag) throws IOException
        {
            channel.basicAck(deliveryTag, false);
        }

        /**
         * 未正常消费， 返回 队列
         *
         * @param channel
         * @param deliveryTag
         * @throws IOException
         */
        default void returnMsg(Channel channel, long deliveryTag) throws IOException
        {
            channel.basicNack(deliveryTag, false, true);
        }
    }

    /**
     * @param correlationData 唯一标识，有了这个唯一标识，我们就知道可以确认（失败）哪一条消息了
     * @param ack             是否投递成功
     * @param cause           失败原因
     */
    private final RabbitTemplate.ConfirmCallback confirmCallback = (CorrelationData correlationData, boolean ack, String cause) ->
    {
        String messageId = correlationData.getId();

        BrokerMessageLogExample brokerMessageLogExample = new BrokerMessageLogExample();
        brokerMessageLogExample.createCriteria().andMessage_idEqualTo(messageId);

        BrokerMessageLog brokerMessageLog = null;
        try
        {
            brokerMessageLog = brokerMessageLogMapper.selectByExample(brokerMessageLogExample).get(0);
        }
        catch (IndexOutOfBoundsException e)
        {
            LOGGER.cody().error("不存在messageId:{}的日志记录", messageId);
        }
        //返回成功，表示消息被正常投递
        if (ack)
        {
            brokerMessageLog.setStatus(MsgStatus.SUCCESS.name());
            brokerMessageLog.setUpdate_time(new Date());
            brokerMessageLogMapper.updateByPrimaryKeySelective(brokerMessageLog);

            LOGGER.cody().info("信息投递成功，messageId:{}", brokerMessageLog.getMessage_id());

        }
        else
        {
            LOGGER.cody().error("消费信息失败，messageId:{} 原因:{}", brokerMessageLog.getMessage_id(), cause);
        }
    };


    @Data
    @Accessors(chain = true)
    public static final class QueueMsgEntity implements Serializable
    {

        private static final long serialVersionUID = 3955200700808952235L;

        private String method;
        private String id;
        private Object object;

    }

    public enum MsgStatus
    {
        SENDING(0, "发送中"), SUCCESS(1, "成功"), FAILURE(2, "失败");

        private final int value;
        private final String message;

        public int value()
        {
            return value;
        }

        public String message()
        {
            return message;
        }

        MsgStatus(int value, String message)
        {

            this.value = value;
            this.message = message;
        }
    }


}
